package com.loginfail_detect

import com.loginfail_detect.bean.{LoginEvent, LoginFailWarning}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                统计2秒钟连续登录失败次数，并输出
  *                (缺陷：无法处理迟到登录数据)
  */
object LoginFail {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val filePath: String = "LoginFailDetect/src/main/resources/LoginLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    //转换成样例类类型，并指定时间戳和watermark
    val loginEventStream: DataStream[LoginEvent] = fileStream.map(data => {
      val arr: Array[String] = data.split(",")
      LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(t: LoginEvent): Long = t.timestamp * 1000L
      })
    //进行判断和检测，如果超过2s之内连续登录失败，输出报警信息
    val loginFailWarningStream: DataStream[LoginFailWarning] = loginEventStream.keyBy(_.userId)
      .process(new LoginFailWarningResult(2))

    loginFailWarningStream.print("fail")

    env.execute("login fail delete job")

  }

  class LoginFailWarningResult(failTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, LoginFailWarning] {
    //定义状态，保存当前所有的登录失败事件，保存定时器的时间戳
    lazy val loginFailListState: ListState[LoginEvent] = getRuntimeContext
      .getListState(new ListStateDescriptor[LoginEvent]("loginfail-list", classOf[LoginEvent]))

    lazy val timerTsState: ValueState[Long] = getRuntimeContext
      .getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))

    override def processElement(i: LoginEvent, context: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#Context, collector: Collector[LoginFailWarning]): Unit = {
      //判断当前登录事件是否成功还是失败
      if (i.eventType == "fail") {
        loginFailListState.add(i)
        //如果没有定时器，那么注册一个2s后的定时器
        if (timerTsState.value() == 0) {
          val ts = i.timestamp * 1000L + 2000L
          context.timerService().registerEventTimeTimer(ts)
          timerTsState.update(ts)
        }
      } else {
        //如果是成功，那么直接清空状态和定时器，重新开始
        context.timerService().deleteEventTimeTimer(timerTsState.value())
        loginFailListState.clear()
        timerTsState.clear()
      }
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#OnTimerContext, out: Collector[LoginFailWarning]): Unit = {
      val allLoginFailList: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]()
      val iter = loginFailListState.get().iterator()
      while (iter.hasNext) {
        allLoginFailList += iter.next()
      }
      //判断登录失败事件个数，如果超过了上限，报警
      if (allLoginFailList.length >= failTimes) {
        println(iter)
        out.collect(LoginFailWarning(
          allLoginFailList.head.userId,
          allLoginFailList.head.timestamp,
          allLoginFailList.last.timestamp,
          "login fail in 2s for " + allLoginFailList.length + " times."))
      }
      //清空状态
      loginFailListState.clear()
      timerTsState.clear()

    }
  }


}
